#include "rpc/net/module/epoll.h"
#include "rpc/net/module/poll.h"
#include "rpc/net/module/select.h"
#include "rpc/net/module/kqueue.h"

#ifdef __linux__
#include <sys/eventfd.h>
#endif

namespace {
void HandleEventFd(int32 fd, void* ctx, int32 event);
}

namespace rpc {
class EventManager::EventFd {
  public:
    virtual ~EventFd() {
    }

    virtual bool Init() = 0;

    virtual void Handle() =0;
    virtual void Notify() = 0;

  protected:
    EventFd(EventManager* ev_mgr, Closure* cb)
        : ev_mgr_(ev_mgr), ev_(new Event), cb_(cb) {
      CHECK_NOTNULL(ev_mgr);
      CHECK_NOTNULL(ev_.get());
      CHECK_NOTNULL(cb);
    }

    EventManager* ev_mgr_;
    scoped_ptr<Event> ev_;
    scoped_ptr<Closure> cb_;

  private:

    DISALLOW_COPY_AND_ASSIGN(EventFd);
};

#ifdef __linux__
class LinuxEventFd: public EventManager::EventFd {
  public:
  EventFd(EventManager* ev_mgr, Closure* cb)
  : EventManager::EventFd(ev_mgr, cb), ev_fd_(-1) {
    CHECK_NOTNULL(ev_mgr);
  }
  ~EventFd() {
    if (ev_fd_ != -1) {
      close(ev_fd_);
      ev_fd_ = -1;
    }
  }

  bool Init();

  void Handle() {
    static uint64 i;
    ::read(ev_fd_, &i, sizeof(i));
    cb_->Run();
  }
  void Notify() {
    const static uint64 i = 1;
    ::write(ev_fd_, &i, sizeof(i));
  }

  private:
  int32 ev_fd_;

  DISALLOW_COPY_AND_ASSIGN(EventFd);
};

bool LinuxEventFd::Init() {
  ev_fd_ = ::eventfd(0, EFD_CLOEXEC);  // block
  if (ev_fd_ == -1) {
    PLOG(WARNING)<< "event_fd error";
    return false;
  }

  ev_.reset(new Event);
  ev_->fd = ev_fd_;
  ev_->arg = this;
  ev_->cb = HandleEventFd;
  ev_->event = EV_READ;

  return ev_mgr_->Add(ev_.get(), EV_READ);
}
#else
class SocketEventFd : public EventManager::EventFd {
  public:
    SocketEventFd(EventManager* ev_mgr, Closure* cb)
        : EventManager::EventFd(ev_mgr, cb) {
      CHECK_NOTNULL(ev_mgr);
      fds_[0] = -1;
      fds_[1] = -1;
    }
    ~SocketEventFd() {
      if (fds_[0] != -1) {
        ::close(fds_[0]);
      }
      if (fds_[1] != -1) {
        ::close(fds_[1]);
      }
    }

    bool Init();

    void Handle() {
      static char c;
      ::read(fds_[0], &c, sizeof(c));
      cb_->Run();
    }
    void Notify() {
      const static char c;
      ::write(fds_[1], &c, sizeof(c));
    }

  private:
    int32 fds_[2];

    DISALLOW_COPY_AND_ASSIGN(SocketEventFd);
};

bool SocketEventFd::Init() {
  if (fds_[0] != -1) return true;
  int32 ret = ::socketpair(AF_LOCAL, SOCK_STREAM, 0, fds_);
  if (ret != 0) {
    LOG(WARNING)<< "socketpair error";
    return false;
  }

  ::shutdown(fds_[0], SHUT_WR);
  ::shutdown(fds_[1], SHUT_RD);

  ev_.reset(new Event);
  ev_->fd = fds_[0];
  ev_->arg = this;
  ev_->cb = HandleEventFd;
  ev_->event = EV_READ;

  return ev_mgr_->Add(ev_.get(), EV_READ);

  return true;
}
#endif

EventManager* CreatePoller() {
  const char* poll = ::getenv("NET_POLL");
  if (poll == NULL) {
#ifdef __linux__
    LOG(INFO)<< "poller use: epoll";
    return new Epoll;
#elif defined (__APPLE__)
    LOG(INFO)<< "poller use kqueue";
    return new KQueue;
#endif

  } else if (::strcmp(poll, "poll") == 0) {
    LOG(INFO)<< "poller use: poll";
    return new Poll;
  }

  LOG(INFO)<< "poller use select";
  return new Select;
}

EventManager::EventManager()
    : exit_(false) {
  loop_tid_ = -1;
  Closure* ev_cb = NewPermanentCallback(this, &EventManager::HandleCb);
  CHECK_NOTNULL(ev_cb);
#ifdef __linux__
  cb_fd_.reset(new LinuxEventFd(this, ev_cb));
#else
  cb_fd_.reset(new SocketEventFd(this, ev_cb));
#endif
  CHECK_NOTNULL(cb_fd_.get());
  CHECK(cb_fd_->Init());
}

void EventManager::RunInLoopThread(Closure* cb) {
  if (::pthread_self() == loop_tid_) {
    AutoRunner(cb);
    return;
  }

  {
    ScopedMutex l(&mutex_);
    tasks_.push_back(cb);
  }
  cb_fd_->Notify();
}

void EventManager::HandleCb() {
  AssertInLoopThread();
  std::deque<Closure*> cbs;
  {
    ScopedMutex l(&mutex_);
    cbs.swap(tasks_);
  }

  while (!cbs.empty()) {
    Closure* c = cbs.front();
    cbs.pop_front();
    c->Run();
  }
}

}

namespace {
void HandleEventFd(int32 fd, void* ctx, int32 event) {
  rpc::EventManager::EventFd* ev = (rpc::EventManager::EventFd*) ctx;
  ev->Handle();
}
}
